﻿using Confluent.Kafka;
using KafkaUtil.Integration;
using KafkaUtil.Interface;
using System;
using System.Linq;
using System.Threading.Tasks;

namespace KafkaUtil.Consumers
{
    public class DefaultKafkaConsumerProvider : IKafkaConsumerProvider
    {
        KafkaConsumerOptions kafkaConsumerOptions;

        public DefaultKafkaConsumerProvider(KafkaConsumerOptions kafkaConsumerOptions)
        {
            this.kafkaConsumerOptions = kafkaConsumerOptions;

            Consumer = KafkaConsumer.Create(kafkaConsumerOptions.GroupId, kafkaConsumerOptions);
            ConsumerOffset = KafkaConsumer.Create(kafkaConsumerOptions.GroupId, kafkaConsumerOptions);
            Consumer.EnableAutoCommit = kafkaConsumerOptions.EnableAutoCommit;
        }

        public KafkaConsumer Consumer { get; }
        /*
         * 新增用于定时读取最新偏移量
         */
        public KafkaConsumer ConsumerOffset { get; }

        public event Action<RecieveResult> OnMessageRecieved;

        public void Dispose()
        {
            Consumer?.Dispose();
            ConsumerOffset?.Dispose();
        }

        /*
         * 20210408 杨世旺新增
         * 获取最新的最大、最小偏移量
         * 会消耗一定资源，请勿频繁调用
         */
        public async Task QueryWatermarkOffsetsAsync(TopicPartition topicPartition, int gap = 1000,  Action<TopicPartition, WatermarkOffsets> action = null)
        {
            await ConsumerOffset.QueryWatermarkOffsetsAsync(kafkaConsumerOptions.Subscribers, topicPartition, gap, WatermarkOffsets =>
            {
                action?.Invoke(topicPartition, WatermarkOffsets);
            });
        }

        public async Task ListenAsync()
        {
            await Consumer.ListenAsync(kafkaConsumerOptions.Subscribers, recieveResult =>
            {
                OnMessageRecieved?.Invoke(recieveResult);
            });
        }
        public override string ToString()
        {
            return $"Consumer-{Consumer}    Subscribers:{string.Join(",", kafkaConsumerOptions.Subscribers.Select(f => f.Partition == null ? f.Topic : $"{f.Topic}:{f.Partition}"))}";
        }
    }
}
